void callback(char *topic, byte *payload, unsigned int length)
{
    Serial.print("Message arrived ");
    //Serial.print(topic);
    //Serial.print("  ");
    String strRcv[2] = {""};
    int i, j;
    j = 0;
    for (i = 0; i < length; i++)
    {
        if ((char)payload[i] == ':')
            j++;
        else
            strRcv[j] += (char)payload[i];
    }
    // Serial.println(strRcv[0]);
    //Serial.println(strRcv[1]);
    int length2 = sizeof(data_stream_name_rw) / sizeof(data_stream_name_rw[0]);
    char tmp[20];
    for (i = 0; i < length2; i++)
    {
        if (strRcv[0] == data_stream_name_rw[i])
        {
            data_stream_value_rw[i] = strRcv[1].toInt();
            Serial.println(data_stream_name_rw[i]);
            Serial.println(data_stream_value_rw[i]);

            snprintf(tmp, sizeof(tmp), "{\"%s\":%d}",
                     data_stream_name_rw[i], data_stream_value_rw[i]);
            pack_msg(tmp);
        }
    }
}
void reconnect()
{
    // Loop until we're reconnected
    if (!client.connected())
    {
        display.clearDisplay();
        display.setTextSize(1);      // Normal 1:1 pixel scale
        display.setTextColor(WHITE); // Draw white text
        display.setCursor(0, 0);
        display.println("Attempting MQTT...");
        display.display();
        Serial.print("Attempting MQTT connection...");
        // Create a random client ID
        String clientId = "ESP8266Client-";
        clientId += String(random(0xffff), HEX);
        // Attempt to connect
        if (client.connect(DeviceID, ProductID, AuthInfo))
        {
            display.clearDisplay();
            display.setTextSize(1);      // Normal 1:1 pixel scale
            display.setTextColor(WHITE); // Draw white text
            display.setCursor(0, 0);
            display.println("connected");
            display.display();

            Serial.println("connected");
            // Once connected, publish an announcement...
            client.publish("outTopic", "hello world");
            // ... and resubscribe
            client.subscribe("inTopic");
        }
        else
        {
            Serial.print("failed, rc=");
            Serial.print(client.state());
            Serial.println(" try again in 5 seconds");
            // Wait 5 seconds before retrying
            // delay(5000);
        }
    }
}
void pack_msg(char *in)
{
    uint16_t streamLen = strlen(in);
    char msg[250];
    char d[3];
    d[0] = '\x03';
    d[1] = (streamLen >> 8);
    d[2] = (streamLen & 0xFF);
    snprintf(msg, sizeof(msg), "%c%c%c%s", d[0], d[1], d[2], in);
    if (client.connected())
    {
        client.publish("$dp", (uint8_t *)msg, streamLen + 3, false);
    }
}
